package org.weishe.pagerank.hbase;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.util.StringUtils;

public class PRMapper extends Mapper<Text, Text, Text, Text> {
	private Text okey = new Text();
	private Text ovalue = new Text();
	private double weight;// 权重

	private String t = "";// 特殊位置

	@Override
	protected void map(Text key, Text value, Mapper<Text, Text, Text, Text>.Context context)
			throws IOException, InterruptedException {

		t = "";
		String[] str = StringUtils.split(value.toString(), ',');
		double w = Double.parseDouble(str[0]);// 得到当前初始权重
		if (str.length > 1) {
			weight = w / (str.length - 1);// 得到每人所得权重
			// 输出各自权重
			for (int i = 1; i < str.length; i++) {
				okey.set(str[i]);
				ovalue.set(key.toString() + "," + weight);
				context.write(okey, ovalue);
				t = t + "," + str[i];
			}
		}
		// 输出特殊位置

		ovalue.set("," + w + t);
		context.write(key, ovalue);
	}

}
